package com.jackstraw.jack_base.mq.rocketmq.config;


import com.jackstraw.jack_base.mq.common.config.AbstractMQConfiguration;
import com.jackstraw.jack_base.mq.rocketmq.aspect.annotation.RocketMQMessageListener;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMQRoutingTemplate;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMQTemplateS;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqConsumer;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqEntity;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqProducer;
import com.jackstraw.jack_base.mq.rocketmq.bean.RocketMqProperties;
import com.jackstraw.jack_base.mq.rocketmq.bean.listener.ProducerRocketMQLocalTransactionListener;
import com.jackstraw.jack_base.mq.rocketmq.bean.listener.ProducerTransactionListener;
import com.jackstraw.jack_base.util.CollectionUtil;
import com.jackstraw.jack_base.util.ObjectUtil;
import com.jackstraw.jack_base.util.StringUtil;
import com.jackstraw.jack_base.util.IPUtil;
import com.jackstraw.jack_base.util.SpringUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.AccessChannel;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.autoconfigure.RocketMQAutoConfiguration;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQReplyListener;
import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.aop.scope.ScopedProxyUtils;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.beans.factory.support.BeanDefinitionValidationException;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.context.support.GenericApplicationContext;
import org.springframework.core.annotation.Order;
import org.springframework.core.env.Environment;

import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.stream.Collectors;


@Slf4j
@Configuration
@EnableAutoConfiguration(exclude = {RocketMQAutoConfiguration.class})
public class RocketMQConfiguration
        extends AbstractMQConfiguration<RocketMqProperties,
        RocketMqEntity,
        RocketMqProducer,
        RocketMqConsumer,
        RocketMQTemplateS,
        RocketMQRoutingTemplate> {

    public RocketMQConfiguration(ApplicationContext applicationContext, Environment environment) {
        super(applicationContext, environment);
    }


    @Bean
    @Primary
    public RocketMQMessageConverter rocketMQMessageConverter() {
        return new RocketMQMessageConverter();
    }

    @Bean
    @Primary
    public TransactionListener transactionListener() {
        return new ProducerTransactionListener();
    }

    @Bean
    @Primary
    public RocketMQLocalTransactionListener rocketMQLocalTransactionListener() {
        return new ProducerRocketMQLocalTransactionListener();
    }

    @Override
    @Bean(value = "rocketMqProperties")
    @ConfigurationProperties(prefix = "jack.mq.rocketmq")
    public RocketMqProperties mqProperties() {
        return new RocketMqProperties();
    }

    @Override
    @Bean(value = "rocketMQRoutingTemplate")
    @ConditionalOnBean(name = {"rocketMqProperties", "rocketMQMessageConverter", "transactionListener", "rocketMQLocalTransactionListener"})
    public RocketMQRoutingTemplate mqRoutingTemplate() {
        RocketMQMessageConverter rocketMQMessageConverter = SpringUtil.getBean("rocketMQMessageConverter", RocketMQMessageConverter.class);
        if (ObjectUtil.ObjectUtilByLang3.isEmpty(rocketMQMessageConverter)) {
            log.error("Bean[rocketMQMessageConverter]注入异常");
            throw new IllegalStateException("Bean[rocketMQMessageConverter]注入异常");
        }

        RocketMqProperties rocketMqProperties = SpringUtil.getBean("rocketMqProperties", RocketMqProperties.class);
        if (ObjectUtil.ObjectUtilByLang3.isEmpty(rocketMqProperties)) {
            log.error("rocketMq配置文件异常");
            throw new IllegalStateException("rocketMq配置文件异常");
        }

        if (ObjectUtil.ObjectUtilByLang3.isEmpty(rocketMqProperties.getMqEntity())) {
            log.error("rocketMq配置文件异常");
            throw new IllegalStateException("rocketMq配置文件异常");
        }

        RocketMQRoutingTemplate rocketMQRoutingTemplate = new RocketMQRoutingTemplate();
        rocketMQRoutingTemplate.setDefaultProperties(rocketMqProperties);
        rocketMQRoutingTemplate.setDefaultServerName(rocketMqProperties.getMqEntity().getServerName());
        rocketMQRoutingTemplate.setDefaultGroupName(rocketMqProperties.getMqEntity().getProducer().getGroup());
        rocketMQRoutingTemplate.setDefaultTopicName(rocketMqProperties.getMqEntity().getConsumer().getTopic());

        Map<String, RocketMQTemplateS> targetMQTemplateSMap = new HashMap<>();
        rocketMQRoutingTemplate.setTargetMQTemplateSMap(targetMQTemplateSMap);

        RocketMQTemplateS rocketMQTemplateS = rocketMqProperties.getMqEntity().mqTemplateS();

        targetMQTemplateSMap.put(rocketMqProperties.getMqEntity().getServerName(), rocketMQTemplateS);
        rocketMQRoutingTemplate.setDefaultMQTemplateS(rocketMQTemplateS);

        return rocketMQRoutingTemplate;
    }

    private AtomicLong counter = new AtomicLong(0);

    @Override
    public void afterSingletonsInstantiated() {
        Map<String, Object> rocketMQMessageListenerBeanS = this.getApplicationContext().getBeansWithAnnotation(RocketMQMessageListener.class)
                .entrySet().stream().filter(entry -> !ScopedProxyUtils.isScopedTarget(entry.getKey()))
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

        rocketMQMessageListenerBeanS.forEach(this::registerContainer);
    }

    private void registerContainer(String beanName, Object bean) {
        Class<?> clazz = AopProxyUtils.ultimateTargetClass(bean);

        if (RocketMQListener.class.isAssignableFrom(bean.getClass()) && RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " cannot be both instance of " + RocketMQListener.class.getName() + " and " + RocketMQReplyListener.class.getName());
        }

        if (!RocketMQListener.class.isAssignableFrom(bean.getClass()) && !RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            throw new IllegalStateException(clazz + " is not instance of " + RocketMQListener.class.getName() + " or " + RocketMQReplyListener.class.getName());
        }

        //取到路由类
        RocketMQRoutingTemplate rocketMQRoutingTemplate = SpringUtil.getBean("rocketMQRoutingTemplate", RocketMQRoutingTemplate.class);

        RocketMQMessageListener annotation = clazz.getAnnotation(RocketMQMessageListener.class);
        String serverName = this.getEnvironment().resolvePlaceholders(annotation.serverName());
        if (StringUtil.StringUtilsBylang3.isEmpty(serverName)) {
            serverName = rocketMQRoutingTemplate.getResolvedServerName();
        }
        RocketMQTemplateS rocketMQTemplateS = rocketMQRoutingTemplate.getResolvedMQTemplateS(serverName);

        String consumerGroup = this.getEnvironment().resolvePlaceholders(annotation.consumerGroup());

        String topic = this.getEnvironment().resolvePlaceholders(annotation.topic());

        AtomicBoolean listenerEnabled =
                new AtomicBoolean((boolean) rocketMQTemplateS.getMqEntity().getConsumer().getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                        .getOrDefault(topic, true));
        List<RocketMqConsumer> otherConsumerS = rocketMQTemplateS.getMqEntity().getOtherConsumerS();
        if (CollectionUtil.isNotEmpty(otherConsumerS)) {
            otherConsumerS.forEach(otherConsumer -> {
                if (ObjectUtil.ObjectUtilByLang3.isNotEmpty(otherConsumer)) {
                    listenerEnabled.set((boolean) otherConsumer.getListeners().getOrDefault(consumerGroup, Collections.EMPTY_MAP)
                            .getOrDefault(topic, true));
                }
            });
        }

        if (!listenerEnabled.get()) {
            log.debug(
                    "Consumer Listener (group:{},topic:{}) is not enabled by configuration, will ignore initialization.",
                    consumerGroup, topic);
            return;
        }
        validate(annotation);

        String containerBeanName = String.format("%s_%s", DefaultRocketMQListenerContainer.class.getName(),
                counter.incrementAndGet());

        GenericApplicationContext genericApplicationContext = (GenericApplicationContext) SpringUtil.getApplicationContext();

        genericApplicationContext.registerBean(containerBeanName, DefaultRocketMQListenerContainer.class,
                () -> createRocketMQListenerContainer(containerBeanName, bean, annotation));
        DefaultRocketMQListenerContainer container = genericApplicationContext.getBean(containerBeanName,
                DefaultRocketMQListenerContainer.class);
        if (!container.isRunning()) {
            try {
                container.start();
            } catch (Exception e) {
                log.error("Started container failed. {}", container, e);
                throw new RuntimeException(e);
            }
        }

        log.info("Register the listener to container, listenerBeanName:{}, containerBeanName:{}", beanName, containerBeanName);
    }

    private DefaultRocketMQListenerContainer createRocketMQListenerContainer(String name, Object bean,
                                                                             RocketMQMessageListener annotation) {
        DefaultRocketMQListenerContainer container = new DefaultRocketMQListenerContainer();

        container.setRocketMQMessageListener(annotation);
        //解析
        container.setSelectorExpression(this.getEnvironment().resolvePlaceholders(annotation.selectorExpression()));
        //取到路由类
        RocketMQRoutingTemplate rocketMQRoutingTemplate = SpringUtil.getBean("rocketMQRoutingTemplate", RocketMQRoutingTemplate.class);
        String serverName = this.getEnvironment().resolvePlaceholders(annotation.serverName());
        if (StringUtil.StringUtilsBylang3.isEmpty(serverName)) {
            serverName = rocketMQRoutingTemplate.getResolvedServerName();
        }
        RocketMQTemplateS rocketMQTemplateS = rocketMQRoutingTemplate.getResolvedMQTemplateS(serverName);
        String nameServer = IPUtil.listToStr(rocketMQTemplateS.getMqEntity().getHostPortS(), ";");
        String accessChannel = this.getEnvironment().resolvePlaceholders(annotation.accessChannel());
        container.setNameServer(nameServer);
        if (!StringUtil.StringUtilsBylang3.isEmpty(accessChannel)) {
            container.setAccessChannel(AccessChannel.valueOf(accessChannel));
        }
        container.setTopic(this.getEnvironment().resolvePlaceholders(annotation.topic()));
        String tags = this.getEnvironment().resolvePlaceholders(annotation.selectorExpression());
        if (!StringUtil.StringUtilsBylang3.isEmpty(tags)) {
            container.setSelectorExpression(tags);
        }

        String consumerGroup = this.getEnvironment().resolvePlaceholders(annotation.consumerGroup());
        if (StringUtil.StringUtilsBylang3.isNotEmpty(serverName)) {
            consumerGroup = consumerGroup.concat("_").concat(serverName);
        }
        container.setConsumerGroup(consumerGroup);
        if (RocketMQListener.class.isAssignableFrom(bean.getClass())) {
            container.setRocketMQListener((RocketMQListener) bean);
        } else if (RocketMQReplyListener.class.isAssignableFrom(bean.getClass())) {
            container.setRocketMQReplyListener((RocketMQReplyListener) bean);
        }
        RocketMQMessageConverter rocketMQMessageConverter = SpringUtil.getBean("rocketMQMessageConverter", RocketMQMessageConverter.class);
        container.setMessageConverter(rocketMQMessageConverter.getMessageConverter());
        container.setName(name);

        return container;
    }

    private void validate(RocketMQMessageListener annotation) {
        if (annotation.consumeMode() == ConsumeMode.ORDERLY &&
                annotation.messageModel() == MessageModel.BROADCASTING) {
            throw new BeanDefinitionValidationException(
                    "Bad annotation definition in @RocketMQMessageListener, messageModel BROADCASTING does not support ORDERLY message!");
        }
    }
}
